import java.text.SimpleDateFormat
import java.util.Locale
import java.util.concurrent.TimeUnit


import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

import scala.io.Source


object TaxiTest {

  def main(args: Array[String]): Unit = {
    // 1. 创建 SparkSession
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("taxi")
      .getOrCreate()

    // 2. 导入函数和隐式转换
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 3. 读取文件
    val taxiRaw: DataFrame = spark.read
      .option("header", value = true)
      .csv("sparktaix/dataset/half_trip.csv")
    //    taxiRaw.show()
    //    taxiRaw.printSchema()

    //4.转换操作
    val taxiParsed: RDD[Either[Trip, (Row, Exception)]] = taxiRaw.rdd.map(safe(parse))
    //可以通过如下方式过滤出来所有异常的row
    //taxiParsed.filter( e => e.isRight)
    //  .map( e => e.right.get._1)
    val taxiGood: Dataset[Trip] = taxiParsed.map( either => either.left.get).toDS()

    //5.绘制时长直方图
    //5.1编写UDF 完成时长计算，将毫秒转为小时单位
    val hours = (pickUpTime: Long, dropOffTime: Long) => {
      val duration = dropOffTime - pickUpTime
      val hours = TimeUnit.HOURS.convert(duration,TimeUnit.MILLISECONDS)
      hours
    }
    val hoursUDF = udf(hours)
    //5.2进行统计
    taxiGood.groupBy(hoursUDF($"pickUpTime",$"dropOffTime") as "duration")
      .count()
      .sort("duration")
      .show()

    //6.根据直方图的显示，查看数据分布后，剪除反常数据
    spark.udf.register("hours",hours)
    val taxiClean = taxiGood.where("hours(pickUpTime,dropOffTime) BETWEEN 0 AND 3")
    //    taxiClean.show()

    //7.5 统计信息
    //    val boroughUDF = udf(boroughLookUp)
    //    taxiClean.groupBy(boroughUDF('dropOffX,'dropOffY))
    //      .count()
    //      .show()

    //8.1 过滤没有经纬度的数据
    //8.2 会话分析
    val session = taxiClean.where("dropOffX != 0 and dropOffY != 0 and pickUpX != 0 and pickUpY != 0")
      .repartition('license)
      .sortWithinPartitions('license,'pickUpTime)

  }

  /**
    * 作用就是封装parse方法，捕获异常
    */
  def safe[P,R](f: P => R): P => Either[R,(P,Exception)]  = {
    new Function[P,Either[R,(P,Exception)]] with Serializable {
      override def apply(param: P): Either[R, (P, Exception)] = {
        try{
          Left(f(param))
        } catch {
          case e: Exception => Right((param,e))
        }
      }
    }
  }

  /**
    * 求上车时间和下车时间的差值
    * 转成小时
    * 行政区查找
    * 有类型的操作用DataSet,无类型的用DataFrame
    * 这里使用DataSet,使用样例类创建Trip对象
    */

  /**
    * 将row转成Trip
    * @param row
    * @return
    */
  def parse(row:Row): Trip = {
    val richRow = new RichRow(row)
    val license = richRow.getAs[String]("hack_license").orNull
    val pickUpTime = parseTime(richRow, "pickup_datetime")
    val dropOffTime = parseTime(richRow, "dropoff_datetime")
    val pickUpX = parseLocation(richRow, "pickup_longitude")
    val pickUpY = parseLocation(richRow, "pickup_latitude")
    val dropOffX = parseLocation(richRow, "dropoff_longitude")
    val dropOffY = parseLocation(richRow, "dropoff_latitude")
    Trip(license,pickUpTime,dropOffTime,pickUpX,pickUpY,dropOffX,dropOffY)
  }

  def parseTime(row: RichRow, field: String): Long = {
    //1.表示出来时间类型的格式 SimpleDateFormat
    val pattern = "yyyy-MM-dd HH:mm:ss"
    val formatter = new SimpleDateFormat(pattern,Locale.ENGLISH)
    //2.执行转换，获取Date对象，getTime获取时间戳
    val time: Option[String] = row.getAs[String](field)
    val timeOption: Option[Long] = time.map(time => formatter.parse(time).getTime)
    timeOption.getOrElse(0L)
  }

  def parseLocation(row: RichRow, field: String): Double = {
    //1.获取数据
    val location = row.getAs[String](field)
    //2.转换数据
    val locationOption = location.map( loc => loc.toDouble)
    locationOption.getOrElse(0.0D)
  }

}

/**
  * DataFrame中Row的包装类型，主要为了包装getAs方法
  * @param row
  */
class RichRow(row : Row){
  /**
    * 为了返回Option提醒外面处理空值，提供处理方式
    */
  def getAs[T](field: String): Option[T] = {
    //1.判断row.getAs 是否为空，row中对应的field是否为空
    if(row.isNullAt(row.fieldIndex(field))){
      //2.null -> 返回None
      None
    } else {
      //3.not null -> 返回Some
      Some(row.getAs[T](field))
    }
  }
}

/**
  * 代表一个行程, 是集合中的一条记录
  */
case class Trip(
                 license: String,//出租车执照号
                 pickUpTime: Long,//上车时间
                 dropOffTime: Long,//下车时间
                 pickUpX: Double,//上车地点的经度
                 pickUpY: Double,//上车地点的纬度
                 dropOffX: Double,//下车地点的经度
                 dropOffY: Double//下车地点的纬度
               )